package com.shujia.streaming

import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}

object WordCountDemo2 {
  def main(args: Array[String]): Unit = {
    /**
     * Spark core:        SparkContext -> RDD
     * Spark Sql:         SparkSession -> DataFrame
     * Spark Streaming:   StreamingContext -> DStream
     */
    //this(sparkContext: SparkContext, batchDuration: Duration)

    val ss: SparkSession = SparkSession.builder()
      .master("local[*]") // 如果不给线程核数，默认只有1个
      .appName("word count")
      .getOrCreate()

    val sc: SparkContext = ss.sparkContext
    sc.setCheckpointDir("spark/data/checkpoint2")

    //    val conf = new SparkConf()
    //    conf.setMaster("local")
    //    conf.setAppName("word count")
    //    val sc = new SparkContext(conf)

    //传入一个SparkContext对象，以及接收要处理的数据时间段
    val streamingContext = new StreamingContext(sc, Durations.seconds(5L))


    //近实时处理中，数据应当是源源不断的
    //使用端口号的数据，模拟源源不断的数据源
    //监控对应端口号中的数据
    val infoDS: ReceiverInputDStream[String] = streamingContext.socketTextStream("master", 12345)

    // hello world hello java
    //因为DStream中封装的是RDD,所以可以使用RDD中的大部分的算子
    val wordsDS: DStream[String] = infoDS.flatMap(_.split(" "))
    val wordKVDS: DStream[(String, Int)] = wordsDS.map((_, 1))
    //使用spark Streaming中的有状态算子，结合历史批次数据进行计算
    //updateStateByKey[S: ClassTag](updateFunc: (Seq[V], Option[S]) => Option[S])
    //1、ByKey算子，只能作用在KV类的DStream上
    //2、自动根据历史中【上一批次】的数据将相同的键为一组，值封装到一个序列中Seq
    //3、Option
    //4、有状态算子，需要结合checkpoint进行使用，需要将历史批次数据保存在磁盘中
    val resDS: DStream[(String, Int)] = wordKVDS.updateStateByKey((seq1: Seq[Int], value: Option[Int]) => {
      val v: Int = value.getOrElse(0)
      val newSum: Int = seq1.sum + v
      Option(newSum)
    })


    println("-----------------------------------")
    resDS.print()
    println("-----------------------------------")

    /**
     * 额外提交命令，一直启动sparkstreaming程序
     */
    streamingContext.start()
    streamingContext.awaitTermination()
    streamingContext.stop()
  }

}
